package com.whz.mlc.factory;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.CacheLoader;
import com.whz.mlc.cache.local.MlcLocalCacheDelegate;
import com.whz.mlc.cache.local.filter.MlcCacheProxyFilter;
import com.whz.mlc.cache.local.MlcLocalCacheBean;
import com.whz.mlc.cache.local.coffeine.CaffeineBuilder;
import com.whz.mlc.cache.second.MlcSecondCacheBean;
import com.whz.mlc.cache.second.RemoteCache;
import com.whz.mlc.cache.second.redis.RedisRemoteCacheDelegate;
import com.whz.mlc.cache.second.redis.filter.RedisRemoteCacheProxyFilter;
import com.whz.mlc.cache.stat.MlcStatManager;
import com.whz.mlc.cache.local.event.listener.LocalCacheRemovalListener;
import com.whz.mlc.hotkey.HotKeyCounter;
import com.whz.mlc.hotkey.HotKeyProperty;
import com.whz.mlc.hotkey.impl.SlidingWindowCounter;
import com.whz.mlc.provider.MlcCacheDefinition;
import com.whz.mlc.provider.loader.LocalCacheLoader;
import com.whz.mlc.provider.loader.MlcCacheLoaderProvider;
import com.whz.mlc.provider.proxy.MlcProxyFactory;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.Assert;

import java.util.Optional;

/**
 * @author whz
 * 2021/8/30
 */
@Data
@Slf4j
public class MlcCacheFactory {
    // 缓存配置 from spring
    private MlcCacheLoaderProvider mlcCacheLoaderProvider;
    // 统计器 from spring
    private MlcStatManager redisMlcStatManager;

    /**
     * 按照缓存明查找 {@link MlcCacheDefinition}
     *
     * @param cacheName 缓存名称
     * @return {@link MlcCacheDefinition}
     */
    public MlcCacheDefinition getMlcCacheDefinition(String cacheName) {
        assertCacheName(cacheName);
        return mlcCacheLoaderProvider.getMlcCacheDefinition(cacheName);
    }


    /**
     * 创建热点统计 {@link SlidingWindowCounter}
     *
     * @param cacheName 缓存名称
     * @return 基于时间窗的{@link SlidingWindowCounter}
     */
    public HotKeyCounter createHotKeyCounter(String cacheName) {
        assertCacheName(cacheName);

        final MlcCacheDefinition mlcCacheDefinition = getMlcCacheDefinition(cacheName);
        HotKeyProperty hotKeyProperty = mlcCacheDefinition.mlcCacheConfig().getHotKeyProperty();
        if (hotKeyProperty == null) {
            hotKeyProperty = HotKeyProperty.builder().build();
            log.info("没有配置hotConfig,默认:{}", hotKeyProperty);
        }
        return Optional.ofNullable(hotKeyProperty.getHotKeyCounter())
                       .orElse(SlidingWindowCounter.builder()
                                                   .cacheName(mlcCacheDefinition.cacheName())
                                                   .config(hotKeyProperty)
                                                   .build());

    }


    /**
     * 创建远程缓存对象
     *
     * @param cacheName   缓存名称
     * @param aopRequired true:需要aop代理缓存的行为;false:无需代理缓存行为
     * @return 远程缓存
     */
    public RemoteCache createSecondCacheProxy(String cacheName, boolean aopRequired) {
        assertCacheName(cacheName);
        final MlcCacheDefinition mlcCacheDefinition = getMlcCacheDefinition(cacheName);
        // redis cache
        RedisRemoteCacheDelegate redisRemoteCacheDelegate = RedisRemoteCacheDelegate.builder()
                                                                                    .cacheName(mlcCacheDefinition.cacheName())
                                                                                    .config(mlcCacheDefinition
                                                                            .mlcCacheConfig()
                                                                            .getRedisConfig())
                                                                                    .build();

        if (aopRequired) {
            return createRemoteCacheProxy(cacheName, redisRemoteCacheDelegate);
        }
        return redisRemoteCacheDelegate;
    }

    /**
     * 按照缓存名称获取二级缓存对象,没有的创建一个新的
     *
     * @param cacheName 缓存名称
     * @return 二级缓存对象
     */
    public MlcSecondCacheBean getSecondCache(final String cacheName) {
        assertCacheName(cacheName);
        if (MlcCacheHolder.hasSecondCache(cacheName)) {
            return MlcCacheHolder.getSecondCache(cacheName);
        }

        final MlcCacheDefinition mlcCacheDefinition = getMlcCacheDefinition(cacheName);
        HotKeyCounter hotKeyCounter = createHotKeyCounter(cacheName);

        MlcSecondCacheBean mlcSecondCacheBean = new MlcSecondCacheBean(cacheName);
        mlcSecondCacheBean.setMlcCacheDefinition(mlcCacheDefinition);
        mlcSecondCacheBean.setRedisMlcStatManager(redisMlcStatManager);
        mlcSecondCacheBean.setProxy(createSecondCacheProxy(cacheName, true));
        mlcSecondCacheBean.setHotKeyCounter(hotKeyCounter);
        mlcSecondCacheBean.afterPropertiesSet();
        // 缓存起来
        MlcCacheHolder.putSecondCache(cacheName, mlcSecondCacheBean);
        return mlcSecondCacheBean;
    }

    /**
     * 按照名称获取本地缓存对象,该对象已经代理了缓存行为.参见{@link MlcLocalCacheDelegate}和{@link MlcCacheProxyFilter}
     *
     * @param cacheName 缓存名称
     * @return 本地缓存对象{@link MlcCacheDefinition}
     */
    public MlcLocalCacheBean getLocalCache(String cacheName) {
        assertCacheName(cacheName);
        if (MlcCacheHolder.hasLocalCache(cacheName)) {
            return MlcCacheHolder.getLocalCache(cacheName);
        }

        Cache<? super Object, ? super Object> cache = buildLocalCacheFromContext(cacheName);
        MlcLocalCacheBean mlcLocalCacheBean = new MlcLocalCacheBean(cacheName, cache);
        mlcLocalCacheBean.setMlcStatManager(redisMlcStatManager);
        mlcLocalCacheBean.afterPropertiesSet();

        // 保存起来
        MlcCacheHolder.putLocalCache(cacheName, mlcLocalCacheBean);
        return mlcLocalCacheBean;
    }

    private Cache<? super Object, ? super Object> buildLocalCacheFromContext(String cacheName) {
        // 从spring context中查找
        final MlcCacheDefinition mlcCacheDefinition = getMlcCacheDefinition(cacheName);

        return CaffeineBuilder.builder()
                              .cacheLoader(buildCacheLoader(cacheName))
                              .cacheName(cacheName)
                              .removalListener(new LocalCacheRemovalListener(cacheName))
                              .build(mlcCacheDefinition.mlcCacheConfig()
                                                       .getCaffeineConfig());
    }

    private CacheLoader<Object, Object> buildCacheLoader(String cacheName) {
        MlcSecondCacheBean mlcSecondCacheBean = getSecondCache(cacheName);
        return new LocalCacheLoader(mlcSecondCacheBean);
    }

    /**
     * 创建远程缓存代理对象,支持对缓存的行为的拦截处理,发送各类异步消息,见{@link RedisRemoteCacheProxyFilter}
     *
     * @param cacheName        缓存名称
     * @param redisRemoteCache redis 远程缓存
     * @return 被代理的缓存对象
     */
    public RemoteCache createRemoteCacheProxy(final String cacheName, final RemoteCache redisRemoteCache) {
        // 拦截cache的行为, RemoteCache的行为在接口上,所以采用jdk代理
        MlcProxyFactory factory = new MlcProxyFactory(redisRemoteCache, RemoteCache.class);
        factory.addProxyFilters(new RedisRemoteCacheProxyFilter(createHotKeyCounter(cacheName)));

        // 二级缓存代理
        return (RemoteCache) factory.getObject();
    }

    private void assertCacheName(final String cacheName) {
        Assert.hasText(cacheName, "cacheName cannot be blank.");
    }
}